package test.threads.queue.unstressed;

import java.util.concurrent.Executor;

import apple.laf.JRSUIConstants.Size;
import com.lmax.disruptor.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import test.helpers.Config;
import test.threads.queue.stressed.AbstractTask;
import test.threads.queue.common.IUsefulWorkEmulator;
import test.threads.queue.stressed.cpu.LongValueMessage;

import static com.google.common.base.Preconditions.checkArgument;

/**
 * (source) -> destination
 *
 * @author cheremin
 * @since 27.02.12,  15:00
 */
public class DisruptorTask extends AbstractTask<LongValueMessage> {
    private static final Log log = LogFactory.getLog( DisruptorTask.class );

    private final WaitStrategy waitStrategy;
    private final boolean publishInBatch;
    private final int batchSize;
    private final boolean subscribeInBatch;

    public DisruptorTask( final Config config ) throws ClassNotFoundException, InstantiationException, IllegalAccessException {
        super( config );

        this.waitStrategy = config.newInstanceByClass(
                "disruptor.wait-strategy",
                new BusySpinWaitStrategy()
        );

        this.publishInBatch = config.getAsBoolean( "disruptor.publish-in-batch", false );
        if ( this.publishInBatch ) {
            batchSize = config.getAsInt( "disruptor.batch-size", 1 );
        } else {
            batchSize = -1;
        }
        this.subscribeInBatch = config.getAsBoolean( "disruptor.subscribe-in-batch", true );
    }


    private RingBuffer<LongValueMessage> ringBuffer;

    private SequenceBarrier lastSequenceBarrier;

    private final Sequence lastSequence = new Sequence( Sequencer.INITIAL_CURSOR_VALUE );

    @Override
    protected EventEnqueuer _initialize( final Executor threadPool,
                                         final IUsefulWorkEmulator<LongValueMessage>[] emulators,
                                         final EventFactory<LongValueMessage> eventFactory ) {
        final ClaimStrategy claimStrategy = new SingleThreadedClaimStrategy( bufferSize() );

        ringBuffer = new RingBuffer<LongValueMessage>(
                eventFactory,
                claimStrategy,
                waitStrategy
        );

        lastSequenceBarrier = ringBuffer.newBarrier();

        ringBuffer.setGatingSequences( lastSequence );

        return new Enqueuer( eventsInBatch(), batchSize, eventFactory, ringBuffer );
    }

    private long sumBeforeGaps = 0;
    private long sumAfterGaps = 0;
    private int gapsMeasured = 0;

    @Override
    protected BenchmarkResult[] executeImpl( final long millisToBogus,
                                             final long millisToMeasure,
                                             final int measurements ) throws Exception {
        sumBeforeGaps = sumAfterGaps = gapsMeasured = 0;

        final BenchmarkResult[] results = super.executeImpl( millisToBogus, millisToMeasure, measurements );

        System.out.printf(
                "|b-gap|=%.2f, |a-gap|=%.2f, gaps=%d\n",
                ( sumBeforeGaps * 1.0 / gapsMeasured ),
                ( sumAfterGaps * 1.0 / gapsMeasured ),
                gapsMeasured
        );

        return results;
    }

    @Override
    protected long dequeueBatchEvents( final int eventsInBatch ) throws Exception {
        final long seq = lastSequence.get();
        final long maxSequence = seq + eventsInBatch;

        sumBeforeGaps += ringBuffer.getCursor() - seq;

        long sum = 0;
        if ( subscribeInBatch ) {
            long nextSequence = lastSequence.get() + 1L;
            while ( nextSequence <= maxSequence ) {
                final long availableSequence = Math.min(
                        lastSequenceBarrier.waitFor( nextSequence ),
                        maxSequence
                );
                while ( nextSequence <= availableSequence ) {
                    final LongValueMessage event = ringBuffer.get( nextSequence );
                    sum += event.value;
                    nextSequence++;
                }
                lastSequence.set( nextSequence - 1L );
            }
        } else {
            for ( long nextSequence = lastSequence.get() + 1L;
                    nextSequence <= maxSequence;
                    nextSequence++ ) {
                lastSequenceBarrier.waitFor( nextSequence );

                final LongValueMessage event = ringBuffer.get( nextSequence );
                sum += event.value;

                lastSequence.set( nextSequence );
            }
        }

        final long gap = ringBuffer.getCursor() - lastSequence.get();
        sumAfterGaps += gap;
        gapsMeasured++;

        /*if ( gap < 4 ) {
            Thread.yield();
        }*/
        return sum;
    }

    @Override
    public int workersRequired() {
        return 1;
    }

    @Override
    public String toString() {
        return String.format(
                "%s[%d][%s][publish:%s][subscribe:%s]",
                getClass().getSimpleName(),
                bufferSize(),
                waitStrategy.getClass().getSimpleName(),
                ( publishInBatch ) ? ( "batch[" + batchSize + "]" ) : ( "no-batch" ),
                ( subscribeInBatch ) ? ( "batch" ) : ( "no-batch" )
        );
    }

    private class Enqueuer extends EventEnqueuer<LongValueMessage> {
        private final RingBuffer<LongValueMessage> ringBuffer;

        private final BatchDescriptor batch;
        private final int batchesInBatch;

        private Enqueuer( final int eventsInBatch,
                          final int batchSize,
                          final EventFactory<LongValueMessage> factory,
                          final RingBuffer<LongValueMessage> ringBuffer ) {
            super( eventsInBatch, factory );

            this.ringBuffer = ringBuffer;

            checkArgument(
                    eventsInBatch % batchSize == 0,
                    "eventsInBatch(%s) % batchSize(%s) must be 0",
                    eventsInBatch, batchSize
            );
            batch = ringBuffer.newBatchDescriptor( batchSize );

            batchesInBatch = eventsInBatch / batchSize;
        }

        @Override
        protected void enqueueEventsBatch( final int eventsInBatch,
                                           final EventFactory<LongValueMessage> factory ) throws Exception {
            if ( publishInBatch ) {
                int i = 0;
                for ( int batchNo = 0; batchNo < batchesInBatch; batchNo++ ) {
                    ringBuffer.next( batch );
                    for ( int eventNo = 0; eventNo < batch.getSize(); eventNo++ ) {
                        final long sequence = batch.getStart() + eventNo;
                        final LongValueMessage event = ringBuffer.get( sequence );
                        event.value = i;
                        i++;
                    }
                    ringBuffer.publish( batch );
                }
            } else {
                for ( int i = 0; i < eventsInBatch; i++ ) {
                    final long nextSequence = ringBuffer.next();
                    final LongValueMessage event = ringBuffer.get( nextSequence );
                    event.value = i;
                    ringBuffer.publish( nextSequence );
                }
            }
        }
    }
}
